package cn.cuilan.common.pipeline;

import cn.cuilan.common.core.ProtocolMessageListener;
import cn.cuilan.common.entity.ProtocolMessage;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Objects;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * 自定义协议综合处理器
 *
 * @param <M> 私有协议消息实体类
 * @author zhang.yan
 */
@ChannelHandler.Sharable
public class CompositeMessageHandler<M extends ProtocolMessage> extends ChannelHandlerAdapter {

    private final Logger logger = LoggerFactory.getLogger(getClass());

    /**
     * 消息处理器集合
     */
    private final List<ProtocolMessageListener<M>> messageListeners = new CopyOnWriteArrayList<>();

    /**
     * 显示失败或错误
     */
    private final boolean failOnError;

    public CompositeMessageHandler() {
        this(true);
    }

    public CompositeMessageHandler(boolean failOnError) {
        this.failOnError = failOnError;
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof ProtocolMessage) {
            doHandleMessage(ctx, msg);
        }
        super.channelRead(ctx, msg);
    }

    private void doHandleMessage(ChannelHandlerContext ctx, Object msg) {
        M protocolMessage;
        try {
            // 无需校验
            protocolMessage = (M) msg;
        } catch (ClassCastException e) {
            logger.debug("ProtocolMessage subclass {} is not supported by {}. Doing nothing.", msg.getClass(), getClass());
            return;
        }

        boolean applyNextListener = true;
        final int size = messageListeners.size();
        for (int i = 0; applyNextListener && i < size; i++) {
            ProtocolMessageListener<M> messageListener = messageListeners.get(i);
            try {
                if (messageListener.applies(protocolMessage)) {
                    applyNextListener = messageListener.onMessage(ctx, protocolMessage);
                    if (!applyNextListener) {
                        logger.warn("Stopping further procession of message {} after handler {}", protocolMessage, messageListener);
                        break;
                    }
                }
            } catch (Exception e) {
                logger.error("Can't evaluate {}.apply({})", messageListener, protocolMessage.getClass(), e);
                if (failOnError) {
                    throw e;
                }
            }
        }
    }

    public List<ProtocolMessageListener<M>> getMessageListeners() {
        return messageListeners;
    }

    public void addListener(ProtocolMessageListener<M> listener) {
        Objects.requireNonNull(listener, "IsoMessageListener is required");
        messageListeners.add(listener);
    }

    @SafeVarargs
    public final void addListeners(ProtocolMessageListener<M>... listeners) {
        Objects.requireNonNull(listeners, "IsoMessageListeners must not be null");
        for (ProtocolMessageListener<M> listener : listeners) {
            addListener(listener);
        }
    }

    public void removeListener(ProtocolMessageListener<M> listener) {
        messageListeners.remove(listener);
    }
}
